package com.rxyb.ops.filter;

import cn.hutool.json.JSONUtil;
import com.rxyb.rabbit.SendQueue;

import static com.rxyb.rabbit.constants.EventConstant.*;
import static com.rxyb.rabbit.constants.EventTypeConstant.RECORD_RSP_LOG;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR;

import com.rxyb.rabbit.dto.EventDTO;
import com.rxyb.rabbit.dto.LogDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.cloud.gateway.support.BodyInserterContext;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.BodyInserter;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.URI;
import java.util.List;

/**
 * <p>
 * 公共Filter
 * </p>
 *
 * @author kylin
 * @since 2019-11-13
 **/
@Slf4j
@Component
public class CommonFilter implements GlobalFilter, Ordered {
    private static final String REQUEST_TIME = "requestTime";

    private static final String CACHE_REQUEST_BODY_OBJECT_KEY = "cachedRequestBodyObject";
    //
    private final SendQueue sendQueue;

    private final List<HttpMessageReader<?>> messageReaders;

    public CommonFilter(SendQueue sendQueue, List<HttpMessageReader<?>> messageReaders) {
        this.sendQueue = sendQueue;
        this.messageReaders = messageReaders;
    }

    @Override
    public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        exchange.getAttributes().put(REQUEST_TIME, System.currentTimeMillis());
        Long startTime = exchange.getAttribute(REQUEST_TIME);
        // 获取请求的类型
        MediaType mediaType = exchange.getRequest().getHeaders().getContentType();
        if (MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)) {
            // 创建一个 server Request 请求
            ServerRequest serverRequest = ServerRequest.create(exchange, this.messageReaders);
            Mono<String> modfileBody = serverRequest.bodyToMono(String.class)
                    .flatMap(body -> {
                        exchange.getAttributes().put(CACHE_REQUEST_BODY_OBJECT_KEY, body);
                        // 创建一个body
                        return Mono.just(body);
                    });
            BodyInserter bodyInserter = BodyInserters.fromPublisher(modfileBody, String.class);
            HttpHeaders headers = new HttpHeaders();
            headers.putAll(exchange.getRequest().getHeaders());
            //新的内容类型将由bodyInserter计算
            //然后在请求装饰器中设置
            headers.remove(HttpHeaders.CONTENT_LANGUAGE);
            CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange,
                    headers);
            return bodyInserter.insert(outputMessage, new BodyInserterContext())
                    .then(Mono.defer(() -> {
                        ServerHttpRequest decorator = decorate(exchange, headers,
                                outputMessage);
                        return chain.filter(exchange.mutate().request(decorator).build()).then(
                                Mono.fromRunnable(() -> {
                                    HttpStatus statusCode = exchange.getResponse().getStatusCode();
                                    if (statusCode == HttpStatus.OK) {
                                        log.debug("请求响应成功，开始发送事件");
                                        doResponse(exchange);
                                    }
                                    log.debug("请求响应失败，状态码{}", statusCode);
                                }));
                    }));
        }
        return chain.filter(exchange).then(
                Mono.fromRunnable(() -> {
                    HttpStatus statusCode = exchange.getResponse().getStatusCode();
                    if (statusCode == HttpStatus.OK) {
                        doResponse(exchange);
                        log.debug("请求响应成功，开始发送事件");
                    }
                    log.info("地址[{}] => 操作时长[{}]", exchange.getRequest().getPath(), System.currentTimeMillis() - startTime);
                })
        );
    }

    private void doResponse(ServerWebExchange exchange) {
        Long startTime = exchange.getAttribute(REQUEST_TIME);
        URI uri = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
        ServerHttpRequest request = exchange.getRequest();
        HttpMethod method = request.getMethod();
        HttpHeaders headers = request.getHeaders();
        MediaType mediaType = exchange.getRequest().getHeaders().getContentType();
        if (uri != null) {
            String path = uri.getPath();
            String query = uri.getQuery();
            EventDTO eventDTO = new EventDTO();
            List<String> authorization = headers.get("Authorization");
            // 登录获取token
            LogDTO dto = new LogDTO();
            if (authorization != null && authorization.size() > 0) {
                dto.setEventDataToken(authorization.get(0));
            }
            // json请求获取body
            if (MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)) {
                dto.setRequestBody(exchange.getAttribute(CACHE_REQUEST_BODY_OBJECT_KEY));
            }
            dto.setRequestPath(path);
            dto.setRequestMethod(method.name());
            dto.setRequestParam(query);
            eventDTO.setContext(JSONUtil.toJsonStr(dto)).setType(RECORD_RSP_LOG);
            log.debug("发送事件到MQ开始");
            long start = System.currentTimeMillis();
            try {
                eventDTO.setQueueName(MY_LOG);
                sendQueue.send(eventDTO);
                log.debug("发送事件到MQ结束，耗时{}ms", System.currentTimeMillis() - start);
            } catch (Exception e) {
                log.error("访问MQ出错,耗时：" + (System.currentTimeMillis() - start) + "ms", e);
            }

        }
        if (startTime != null) {
            log.info(exchange.getRequest().getURI().getRawPath() + ": " + (System.currentTimeMillis() - startTime) + "ms");
        }
    }

    private ServerHttpRequestDecorator decorate(ServerWebExchange exchange, HttpHeaders headers,
                                                CachedBodyOutputMessage outputMessage) {
        return new ServerHttpRequestDecorator(exchange.getRequest()) {
            @Override
            public HttpHeaders getHeaders() {
                long contentLength = headers.getContentLength();
                HttpHeaders httpHeaders = new HttpHeaders();
                httpHeaders.putAll(super.getHeaders());
                if (contentLength > 0) {
                    httpHeaders.setContentLength(contentLength);
                } else {
                    // TODO: this causes a 'HTTP/1.1 411 Length Required' // on
                    // httpbin.org
                    httpHeaders.set(HttpHeaders.TRANSFER_ENCODING, "chunked");
                }
                return httpHeaders;
            }

            @Override
            public Flux<DataBuffer> getBody() {
                return outputMessage.getBody();
            }
        };
    }


    @Override
    public int getOrder() {
        return -1000;
    }
}
